package com.dt.platform.ops.service.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.dt.platform.constants.enums.ops.MonitorMethodEnum;
import com.dt.platform.domain.ops.MonitorNode;
import com.dt.platform.domain.ops.MonitorTplIndicator;
import com.dt.platform.ops.service.IMonitorDataProcessMagicApiService;
import com.dt.platform.ops.service.IMonitorNodeService;
import com.dt.platform.proxy.common.CommonServiceProxy;
import com.github.foxnic.api.error.ErrorDesc;
import com.github.foxnic.api.transter.Result;
import com.github.foxnic.commons.busi.id.IDGenerator;
import com.github.foxnic.commons.concurrent.SimpleJoinForkTask;
import com.github.foxnic.commons.log.Logger;
import com.github.foxnic.dao.data.RcdSet;
import com.github.foxnic.dao.spec.DAO;
import com.github.foxnic.sql.expr.Insert;
import org.github.foxnic.web.framework.dao.DBConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

@Service("MonitorDataProcessMagicApiService")
public class MonitorDataProcessMagicApiServiceImpl implements IMonitorDataProcessMagicApiService {

    /**
     * 注入DAO对象
     * */
    @Resource(name= DBConfigs.PRIMARY_DAO)
    private DAO dao=null;

    /**
     * 获得 DAO 对象
     * */
    public DAO dao() { return dao; }


    @Autowired
    private MonitorDataProcessBaseServiceImpl monitorDataProcessBaseService;


    @Override
    public Result collectData() {
        List<MonitorNode> nodeList=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.MAGIC_API.code());
        Logger.info("collectData,find node number:"+nodeList.size());
        if(nodeList.size()>0){
            int batchSize=monitorDataProcessBaseService.calBathSizeValue(nodeList.size());
            Logger.info("current batch size:"+batchSize);
            // 创建 ForkJoin 工具，其中 输入一个 Integer 元素的 List ，输出 Long 元素的 List ，每个线程处理 若干元素 ，此处为 5 个
            SimpleJoinForkTask<MonitorNode,Result> task=new SimpleJoinForkTask<>(nodeList,batchSize);
            // 并行执行
            List<Result> rvs=task.execute(els->{
                // 打印当前线程信息
                Logger.info(Thread.currentThread().getName());
                // 处理当前分到的 若干元素，此处为 5 个
                List<Result> rs=new ArrayList<>();
                for (MonitorNode node : els) {
                    Result result=collectNodeData(node,null,false);
                    if(result.isSuccess()){
                        Logger.debug("node:"+node.getNodeNameShow()+",result:"+result.success()+",message:"+result.getMessage());
                    }else{
                        Logger.error("node:"+node.getNodeNameShow()+",result:"+result.success()+",message:"+result.getMessage());
                    }
                    rs.add(result);
                }
                // 处理完毕，返回本批次的处理结果，由 SimpleJoinForkTask 负责合并全部结果，并由 rvs 变量接收
                return rs;
            });
        }
        return ErrorDesc.success();
    }


    @Override
    public Result collectNodeData(MonitorNode node,String indicatorId,boolean isForce) {
        String me="magic_api";
        List<MonitorTplIndicator> monitorTplIndicatorList=monitorDataProcessBaseService.queryExecuteIndicatorList(node.getId(),me,null,isForce);
        if(monitorTplIndicatorList==null||monitorTplIndicatorList.size()==0){
            Logger.info("collect method:magic api,node:"+node.getNodeNameShow()+",当前没有指标触发采集");
            return ErrorDesc.success();
        }
        List<Insert> list=queryIndicators(monitorTplIndicatorList,node);
        if(list!=null&&list.size()>0){
            monitorDataProcessBaseService.executeCollectDataInsert(list,"collect by magic api");
        }
        return ErrorDesc.success();

    }

    private List<Insert> queryIndicators(List<MonitorTplIndicator> monitorTplIndicatorList,MonitorNode node){

        List<Insert> list=new ArrayList<>();
        for(int i=0;i<monitorTplIndicatorList.size();i++){
            MonitorTplIndicator indicator=monitorTplIndicatorList.get(i);
            HashMap<String,Object> ps=new HashMap<>();
            JSONObject nodeObj=new JSONObject();
            nodeObj.put("id",node.getId());
            nodeObj.put("nodeIp",node.getNodeIp());
            ps.put("node",nodeObj.toJSONString());
            ps.put("nodeVar",node.getVar());
            Result<Object> res=CommonServiceProxy.api().callMagicAPIService("execute", "POST", indicator.getCommand(), ps);
            Logger.info("node:"+node.getNodeNameShow()+",indicator:"+indicator.getName()+",res:"+res.getData().toString());
           if(res.success()){
               JSONArray rs=(JSONArray)res.getData();
               if(rs==null||rs.size()==0){
                   Insert valueInsert=new Insert("ops_monitor_node_value");
                   valueInsert.set("id", IDGenerator.getSnowflakeId());
                   valueInsert.setIf("result_status","failed");
                   valueInsert.setIf("result_message","result is null");
                   valueInsert.setIf("indicator_code",indicator.getCode());
                   valueInsert.setIf("node_id",node.getId());
                   valueInsert.setIf("is_connected",1);
                   valueInsert.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                   valueInsert.setIf("record_time",new Date());
                   list.add(valueInsert);
               }else{
                   String[] mapArr=indicator.getValueColumnMap().split(",");
                   String content="";
                   boolean dataCheck=true;
                   String dataCheckMsg="";
                   for(int k=0;k<rs.size();k++){
                       String e="";
                       for(int j=0;j<mapArr.length;j++){
                           String key=mapArr[j];
                           JSONObject v=rs.getJSONObject(k);
                           if(v.containsKey(key)){
                               if(j==0){
                                   e=v.getString(key);
                               }else{
                                   e=e+","+v.getString(key);
                               }
                           }else{
                               dataCheckMsg="column key "+key+" not exist.";
                               Logger.info(dataCheckMsg);
                               dataCheck=false;
                               break;
                           }
                       }
                       if(k==0){
                           content=e;
                       }else{
                           content=content+"\n"+e;
                       }
                   }
                   if(dataCheck){
                       Logger.info("content:\n"+content);
                       Result<Object> insertResult=monitorDataProcessBaseService.convertToInsertData(indicator,content,node);
                       if(insertResult.isSuccess()){
                           list=(List<Insert>)insertResult.getData();
                       }else{
                           Insert valueInsert2=new Insert("ops_monitor_node_value");
                           valueInsert2.set("id", IDGenerator.getSnowflakeId());
                           valueInsert2.setIf("result_status","failed");
                           valueInsert2.setIf("result_message",insertResult.getMessage());
                           valueInsert2.setIf("indicator_code",indicator.getCode());
                           valueInsert2.setIf("node_id",node.getId());
                           valueInsert2.setIf("is_connected",1);
                           valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                           valueInsert2.setIf("record_time",new Date());
                           list.add(valueInsert2);
                       }
                   }else{
                       Insert valueInsert2=new Insert("ops_monitor_node_value");
                       valueInsert2.set("id", IDGenerator.getSnowflakeId());
                       valueInsert2.setIf("result_status","failed");
                       valueInsert2.setIf("result_message",dataCheckMsg);
                       valueInsert2.setIf("indicator_code",indicator.getCode());
                       valueInsert2.setIf("node_id",node.getId());
                       valueInsert2.setIf("is_connected",1);
                       valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                       valueInsert2.setIf("record_time",new Date());
                       list.add(valueInsert2);
                   }
               }
           }else{
               Insert valueInsert2=new Insert("ops_monitor_node_value");
               valueInsert2.set("id", IDGenerator.getSnowflakeId());
               valueInsert2.setIf("result_status","failed");
               valueInsert2.setIf("result_message",res.getMessage().length()>1024?res.getMessage().substring(0,1000):res.getMessage());
               valueInsert2.setIf("indicator_code",indicator.getCode());
               valueInsert2.setIf("node_id",node.getId());
               valueInsert2.setIf("is_connected",1);
               valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
               valueInsert2.setIf("record_time",new Date());
               list.add(valueInsert2);
           }
        }
        return list;
    }

}
